/*
 * Copyright 2007-2008  Lukasz Laszko lukaszlaszko@gmail.com
 * 
 * Permission to use, copy, modify, and distribute this software
 * and its documentation for any purpose and without fee is hereby
 * granted, provided that the above copyright notice appear in all
 * copies and that both that the copyright notice and this
 * permission notice and warranty disclaimer appear in supporting
 * documentation, and that the name of the author not be used in
 * advertising or publicity pertaining to distribution of the
 * software without specific, written prior permission.
 * 
 * The author disclaim all warranties with regard to this
 * software, including all implied warranties of merchantability
 * and fitness.  In no event shall the author be liable for any
 * special, indirect or consequential damages or any damages
 * whatsoever resulting from loss of use, data or profits, whether
 * in an action of contract, negligence or other tortious action,
 * arising out of or in connection with the use or performance of
 * this software. 
 */
package org.pg.biomedics.asr.concurrent;

import java.util.concurrent.Executors;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;

import org.apache.log4j.Logger;
import org.apache.commons.pool.ObjectPool;
import org.apache.commons.pool.PoolableObjectFactory;
import org.apache.commons.pool.impl.StackObjectPool; 

import org.pg.biomedics.asr.model.Task;
import org.pg.biomedics.asr.model.Result;
import org.pg.biomedics.asr.model.code.TaskPriorityCodes;
import org.pg.biomedics.asr.model.code.TaskStateCodes;
import org.pg.biomedics.asr.model.code.ResultTypeCodes;
import org.pg.biomedics.asr.data.TaskManager;
import org.pg.biomedics.asr.data.ResultManager;
import org.pg.biomedics.asr.data.DataAccessException;
import org.pg.biomedics.asr.config.ConfigurationManager;
import org.pg.biomedics.asr.config.ConfigurationException;

/**
 * Task scheduler encapsulates logic for task executing with thread pool
 * @author Lukasz Laszko
 */
public class TaskScheduler 
{
    // <editor-fold desc="Fields">
    
    private Logger logger = Logger.getLogger(TaskScheduler.class);
    
    int maxThreads;    
    String voiceEngineConfig;
    
    private ObjectPool voiceEnginePool;
    
    private ExecutorService executor;
    private CompletionService<Result> completionService;
    
    private TaskManager taskManager;
    private ResultManager resultManager;
    
    private boolean isRunning;       
    
    // </editor-fold>
    
    // <editor-fold desc="Constructors">
    
    /**
     * Creates instance of task scheduler and assigns new voice engine config file location
     * @throws org.pg.biomedics.asr.concurrent.TaskSchedulerException
     */
    public TaskScheduler() throws TaskSchedulerException 
    {
        try
        {
            ConfigurationManager configuration = new ConfigurationManager();
            
            this.maxThreads = configuration.getSchedulerMaxThreads();
            this.voiceEngineConfig = configuration.getVoiceEngineConfig();
            
            this.taskManager = new TaskManager(configuration.getSqlMapConfig());
            this.resultManager = new ResultManager(configuration.getSqlMapConfig());
        }        
        catch (ConfigurationException cex)
        {
            throw new TaskSchedulerException("Can't create task scheduler, configuration error", cex);
        }
        catch (DataAccessException daex)
        {
            throw new TaskSchedulerException("Can't create task scheduler, data access modeuls initalization error", daex);
        }
    }
    
    // <editor-fold>
    
    // <editor-fold desc="Properties">
    
    public boolean isRunning()
    {
        return this.isRunning;
    }
    
    // </editor-fold>
    
    // <editor-fold desc="Methods">
    
    /**
     * Provides initialization logic for scheduler
     * @throws org.pg.biomedics.asr.concurrent.TaskSchedulerException
     */
    public void initialize() throws TaskSchedulerException
    {
        if (!this.isRunning)
        {
            try
            {
                // create engines pool
                this.voiceEnginePool = new StackObjectPool(this.maxThreads, this.maxThreads);
                this.voiceEnginePool.setFactory(new PoolableObjectFactory() 
                {
                    public Object makeObject() throws Exception
                    {
                        VoiceEngine voiceEngine = new VoiceEngine(voiceEngineConfig);
                        voiceEngine.initialize();
                        
                        return voiceEngine;
                    }

                    public void destroyObject(Object obj) throws Exception
                    {
                        VoiceEngine voiceEngine = (VoiceEngine)obj;
                        voiceEngine.shutdown();                                               
                    }

                    public boolean validateObject(Object obj)
                    {
                        VoiceEngine voiceEngine = (VoiceEngine)obj;
                        return voiceEngine.isReady();
                    }

                    public void activateObject(Object obj) throws Exception
                    { }

                    public void passivateObject(Object obj) throws Exception
                    { }
                });

                for (int i = 0; i < this.maxThreads; i++)
                    this.voiceEnginePool.addObject();

                // create executors
                this.executor = Executors.newFixedThreadPool(maxThreads);
                this.completionService = new ExecutorCompletionService(this.executor);               
            }
            catch (Exception ex)
            {
                throw new TaskSchedulerException("Can't initialize task scheduler", ex);
            }
        }
    }
    
    /**
     * Executes scheduler routine
     */
    public void run()
    {        
        this.isRunning = true; 
        
        // fill execution queue with tasks stored in database
        try
        {
            Task[] tasks = loadTasks();
            for (int i = 0; i < tasks.length; i++)
                submit(tasks[i]);
        }
        catch (Exception ex)
        {
            logger.error("Can't insert or get result from completion service", ex);
        }
        
        // wait for job results                       
        while (this.isRunning)
        {
            try
            {
                // get result and notifiy
                Future<Result> resultFuture = this.completionService.take();

                // save results to database
                Result result = resultFuture.get();
                this.resultManager.insertResult(result);                
            }
            catch (InterruptedException iex)
            {
                this.isRunning = false;
            }
            catch (Exception ex)
            {
                logger.error("Can't insert or get result from completion service", ex);
            }
        }
    }
    
    /**
     * Provides deinitialization logic for scheduler
     */
    public void shutdown()
    {
        if (this.isRunning)
        {
            // shutdown tasks
            Future<Result> future = null;
            while ((future = this.completionService.poll()) != null)
                future.cancel(true);
        }
    }
       
    /**
     * Submits a new task for execution
     * @param task task to execute
     * @throws org.pg.biomedics.asr.concurrent.TaskSchedulerException
     */
    public void submit(Task task) throws TaskSchedulerException
    {
        if (!this.isRunning)
            throw new TaskSchedulerException("Scheduler must be initializaed before task submition");
                
        // create new callable and state listener
        VoiceEngineCallable callableTask = new VoiceEngineCallable(task, voiceEnginePool);
        callableTask.addTaskCallListener(new TaskCallListener() 
        {
            public void callExecuting(Task task) throws Exception
            {
                taskManager.SetTaskState(task, TaskStateCodes.PENDING);
            }
    
            public void callExecuted(Task task, Result result) throws Exception
            {
                if (ResultTypeCodes.ERROR.equals(result.getTypeCode()))
                    taskManager.SetTaskState(task, TaskStateCodes.FAILED);                    
                else
                    taskManager.SetTaskState(task, TaskStateCodes.SUCCESS);                    
            }
        });
        
        // submit callable to thread pool
        this.completionService.submit(callableTask);
    }
    
    // </editor-fold>
    
    // <editor-fold desc="Support methods">
    
    /**
     * Loads tasks stored in database in wainting and pending state. This tasks are 
     * sorted according to priority
     * @return list of tasks to execute
     */
    private Task[] loadTasks() throws DataAccessException
    {
        // get tasks from database
        Task[] pendingTasks = this.taskManager.getTasks(TaskStateCodes.PENDING);
        Task[] waitingTasks = this.taskManager.getTasks(TaskStateCodes.WAITING);
        
        Task[] allTasks = new Task[pendingTasks.length + waitingTasks.length]; 
        int index = 0;
        
        // sort pending tasks        
        for (int i = 0; i < pendingTasks.length; i++)
            if (pendingTasks[i].getPriorityCode() != null)
                if (pendingTasks[i].getPriorityCode().equals(TaskPriorityCodes.HIGH))
                    allTasks[index++] = pendingTasks[i];
        
        for (int i = 0; i < pendingTasks.length; i++)
            if (pendingTasks[i].getPriorityCode() != null)
                if (pendingTasks[i].getPriorityCode().equals(TaskPriorityCodes.NORMAL))
                    allTasks[index++] = pendingTasks[i];
        
        for (int i = 0; i < pendingTasks.length; i++)
            if (pendingTasks[i].getPriorityCode() != null)
                if (pendingTasks[i].getPriorityCode().equals(TaskPriorityCodes.LOW))
                    allTasks[index++] = pendingTasks[i];
        
        // sort waiting tasks
        for (int i = 0; i < waitingTasks.length; i++)
            if (waitingTasks[i].getPriorityCode() != null)
                if (waitingTasks[i].getPriorityCode().equals(TaskPriorityCodes.HIGH))
                    allTasks[index++] = waitingTasks[i];
        
        for (int i = 0; i < waitingTasks.length; i++)
            if (waitingTasks[i].getPriorityCode() != null)
                if (waitingTasks[i].getPriorityCode().equals(TaskPriorityCodes.NORMAL))
                    allTasks[index++] = waitingTasks[i];
        
        for (int i = 0; i < waitingTasks.length; i++)
            if (waitingTasks[i].getPriorityCode() != null)
                if (waitingTasks[i].getPriorityCode().equals(TaskPriorityCodes.LOW))
                    allTasks[index++] = waitingTasks[i];
        
        return allTasks;
    }
    
    // </editor-fold>
}
